package com.deng.eventflow.core.tmf;

import com.deng.eventflow.core.Pushable;
import com.deng.eventflow.core.SynchronousProcessor;
import com.deng.eventflow.core.exception.ProcessorException;
import com.deng.eventflow.core.exception.PushableException;

import java.util.ArrayDeque;
import java.util.Queue;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;

public abstract class Source extends SynchronousProcessor {

    public Source(int out_arity)
    {
        super(0, out_arity);
    }

    public final void push() {
        Queue<Object[]> output = new ArrayDeque<Object[]>(1);
        try
        {
            compute(null, output);
        }
        catch (ProcessorException e)
        {
            throw new PushableException(e);
        }
        if (output.isEmpty())
        {
            return;
        }
        for (Object[] evt : output)
        {
            @SuppressWarnings("unchecked")
            Future<Pushable>[] futures = new Future[output.size()];
            if (evt != null && !allNull(evt))
            {
                for (int i = 0; i < output.size(); i++)
                {
                    Pushable p = m_outputPushables[i];
                    futures[i] = p.pushFast(evt[i]);
                }
                for (int i = 0; i < output.size(); i++)
                {
                    try
                    {
                        futures[i].get();
                    }
                    catch (InterruptedException e)
                    {
                        throw new ProcessorException(e);
                    }
                    catch (ExecutionException e)
                    {
                        throw new ProcessorException(e);
                    }
                }
            }
        }
    }
}
